package com.atguigu;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/17 20:20
 */
public class ETDemo1 extends BaseApp {
    public static void main(String[] args) {
        new ETDemo1().init(1, "ETDemo1", "dwd_page_log");
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> sourceStream) {
        setWebUi(env, 3000);
        sourceStream
            .map(line -> {
                final JSONObject obj = JSON.parseObject(line);
                return obj;
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                })
            )
            .process(new ProcessFunction<JSONObject, Object>() {
                @Override
                public void processElement(JSONObject value, Context ctx, Collector<Object> out) throws Exception {
                    System.out.println(ctx.timerService().currentWatermark());
                }
            }).setParallelism(2)
            .print();
    }
}
